package com.audaque.springboot.foshanupload.web.amqpdemo.controller;

import com.audaque.springboot.foshanupload.web.amqpdemo.model.rsp.SysUserMpgtRsp;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

@Slf4j
@RestController
@RequestMapping("/consumerClient")
public class ConsumerClientController {

    @Autowired
    private ConnectionFactory connectionFactory;

    @PostMapping("/queue")
    public void queue( //路由键：队列的key
                            @RequestParam(value = "queue", required = false) String queue,
                            @RequestParam(value = "autoAck", required = false) Boolean autoAck,
                            @RequestBody SysUserMpgtRsp sysUserMpgtRs) throws IOException, TimeoutException {
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(queue, true, false, false, null);
        //qos:在非自动确认消息的前提下，如果一定数目的消息（通过基于 consume 或者 channel 设置 Qos 的值）未被确认前，不进行消费新的消息。
        channel.basicQos(1);//一次只接受一条未确认的消息
        //参数2:关闭自动确认消息
        autoAck = autoAck == null ? false : autoAck;
        Boolean finalAutoAck = autoAck;
        channel.basicConsume(queue, autoAck, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1: " + new String(body));
                if (finalAutoAck) {
                    channel.basicAck(envelope.getDeliveryTag(), false);//手动确认消息
                }
            }
        });


    }

    @PostMapping("/exchange")
    public void exchange(@RequestParam(value = "exchange", required = false) String exchange,
                            @RequestParam(value = "autoAck", required = false) Boolean autoAck,
                            @RequestBody SysUserMpgtRsp sysUserMpgtRs) throws IOException, TimeoutException {
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        //绑定交换机
        channel.exchangeDeclare(exchange,"fanout");
        //创建临时队列
        String queue = channel.queueDeclare().getQueue();
        //将临时队列绑定exchange
        channel.queueBind(queue,exchange,"");


        //qos:在非自动确认消息的前提下，如果一定数目的消息（通过基于 consume 或者 channel 设置 Qos 的值）未被确认前，不进行消费新的消息。
        channel.basicQos(1);//一次只接受一条未确认的消息
        //参数2:关闭自动确认消息
        autoAck = autoAck == null ? false : autoAck;
        Boolean finalAutoAck = autoAck;
        channel.basicConsume(queue, autoAck, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1: " + new String(body));
                if (finalAutoAck) {
                    channel.basicAck(envelope.getDeliveryTag(), false);//手动确认消息
                }
            }
        });

    }


    @PostMapping("/direct")
    public void direct(@RequestParam(value = "exchange", required = false) String exchange,
            //路由键：队列的key
                            @RequestParam(value = "routingKey", required = false) String routingKey,
                            @RequestParam(value = "autoAck", required = false) Boolean autoAck,
                            @RequestBody SysUserMpgtRsp sysUserMpgtRs) throws IOException, TimeoutException {
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        //声明交换机
        channel.exchangeDeclare(exchange,"direct");
        //创建临时队列
        String queue = channel.queueDeclare().getQueue();
        //绑定队列和交换机
        channel.queueBind(queue,exchange,routingKey);



        //qos:在非自动确认消息的前提下，如果一定数目的消息（通过基于 consume 或者 channel 设置 Qos 的值）未被确认前，不进行消费新的消息。
        channel.basicQos(1);//一次只接受一条未确认的消息
        //参数2:关闭自动确认消息
        autoAck = autoAck == null ? false : autoAck;
        Boolean finalAutoAck = autoAck;
        channel.basicConsume(queue, autoAck, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1: " + new String(body));
                if (finalAutoAck) {
                    channel.basicAck(envelope.getDeliveryTag(), false);//手动确认消息
                }
            }
        });

    }



    @PostMapping("/topic")
    public void topic(@RequestParam(value = "exchange", required = false) String exchange,
            //路由键：队列的key
                            @RequestParam(value = "routingKey", required = false) String routingKey,
                            @RequestParam(value = "autoAck", required = false) Boolean autoAck,
                            @RequestBody SysUserMpgtRsp sysUserMpgtRs) throws IOException, TimeoutException {
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        //声明交换机
        channel.exchangeDeclare(exchange,"topic");
        //创建临时队列
        String queue = channel.queueDeclare().getQueue();
        //绑定队列和交换机
        channel.queueBind(queue,exchange,routingKey);



        //qos:在非自动确认消息的前提下，如果一定数目的消息（通过基于 consume 或者 channel 设置 Qos 的值）未被确认前，不进行消费新的消息。
        channel.basicQos(1);//一次只接受一条未确认的消息
        //参数2:关闭自动确认消息
        autoAck = autoAck == null ? false : autoAck;
        Boolean finalAutoAck = autoAck;
        channel.basicConsume(queue, autoAck, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1: " + new String(body));
                if (finalAutoAck) {
                    channel.basicAck(envelope.getDeliveryTag(), false);//手动确认消息
                }
            }
        });

    }






    @PostMapping("/ttl")
    public void ttl(@RequestParam(value = "exchange", required = false) String exchange,
            //路由键：队列的key
                            @RequestParam(value = "routingKey", required = false) String routingKey,
                            @RequestParam(value = "autoAck", required = false) Boolean autoAck,
                            @RequestBody SysUserMpgtRsp sysUserMpgtRs) throws IOException, TimeoutException {
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();


        //必须设置参数到 arguments 中
        Map<String, Object> arguments = new HashMap<String, Object>();
        arguments.put("x-dead-letter-exchange", "dlx.exchange");

        channel.exchangeDeclare(exchange, "topic", true, false, null);
        //将 arguments 放入队列的声明中
        channel.queueDeclare(routingKey, true, false, false, arguments);

        //一般不用代码绑定，在管理界面手动绑定
        channel.queueBind(routingKey, exchange, routingKey);

        //声明死信队列,专用
        channel.exchangeDeclare(exchange.concat("dlx.exchange"), "topic", true, false, null);
        channel.queueDeclare(exchange.concat("dlx.queue"), true, false, false, null);
        //路由键为 # 代表可以路由到所有消息
        channel.queueBind(exchange.concat("dlx.queue"), exchange.concat("dlx.exchange"), "#");



        //qos:在非自动确认消息的前提下，如果一定数目的消息（通过基于 consume 或者 channel 设置 Qos 的值）未被确认前，不进行消费新的消息。
        channel.basicQos(1);//一次只接受一条未确认的消息
        //参数2:关闭自动确认消息
        autoAck = autoAck == null ? false : autoAck;
        Boolean finalAutoAck = autoAck;
        channel.basicConsume(routingKey, autoAck, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1: " + new String(body, "UTF-8"));
                if (finalAutoAck) {
                    channel.basicAck(envelope.getDeliveryTag(), false);//手动确认消息
                }
            }
        });

    }














}
